package ac.il.technion.cs236635.project.core;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URL;
import java.util.Properties;
import java.util.Random;
import java.util.Vector;

import javax.xml.namespace.QName;
import javax.xml.ws.Endpoint;

import org.apache.log4j.Logger;
import org.jgroups.Address;
import org.jgroups.ChannelException;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;
import org.jgroups.util.Util;

import ac.il.technion.cs236635.project.core.dispatcher.DispatcherService;
import ac.il.technion.cs236635.project.core.dispatcher.ServerInfo;
import ac.il.technion.cs236635.project.core.messages.ClusterMessage;
import ac.il.technion.cs236635.project.core.messages.ClusterMessageType;

public class Server extends ReceiverAdapter implements StopListener, Runnable, RejectListener {

	private boolean 	finished 		= false;
	
	private Endpoint 	endpoint 		= null;
	
	private String 		serverName 		= null;
	
	private boolean 	highPriority 	= false;

	private JChannel 	channel			= null;
	
	private Logger 		log				= Logger.getLogger( Server.class);

	
	public Server( final String name, final String serverUrl) {
		this.serverName = name;
		endpoint = Endpoint.create( new CloudWorker( ));
		endpoint.publish( serverUrl);
		// Initialize cluster channel communication
		try {
			channel = new JChannel( );
			channel.setName( serverName);
			channel.connect( "CloudCommunication");
		} catch (ChannelException e) {
			log.error(e);
		}
	}
	
	@Override
	public void receive( Message msg) {
		if ( msg.getSrc( ) == channel.getAddress( ) ) // Ignore self messages
			return;
		try {
			ClusterMessage message = (ClusterMessage)Util.objectFromByteBuffer( msg.getBuffer( ));
			ServerJob job = ( ServerJob)Util.objectFromByteBuffer( message.getAttachement( ));
			switch ( message.getType( )) {
				case JOB_FINISHED : {
					if ( job.getJobType() == JobType.HIGH_PRIORITY) {
						JobsQueue.getInstance().removeLowJobById( job.getJobId( ));
					} else {
						JobsQueue.getInstance().removeHighJobById( job.getJobId( ));
					}
					break;
				}
				case JOB_REDIRECT : {
					if ( job.getJobType() == JobType.HIGH_PRIORITY) {
						JobsQueue.getInstance().pushHighPriorityJob(job);
					} else {
						JobsQueue.getInstance().pushLowPriorityJob(job);
					}
					break;
				}
				default : 
							break;
			}
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void viewAccepted(View new_view) {
		// Everyone need to send his credentials
		super.viewAccepted(new_view);
	}

	@Override
	public void run( ) {
		ServerJob job = null;
		while( !finished) {
			if ( highPriority) {
				// Executing high priority job, cannot preempt.
				// Need to wait until it finishes.
				try {
					Thread.sleep( 100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				continue;
			}
			ServerJob $ = JobsQueue.getInstance().popHighPriorityJob( );
			if ( $ != null ) {
				// There is a new high priority job  arrived, there is a need to 
				// preempt the current low priority job and execute the new one.
				if ( job != null && !job.isFinished( )) {
					// Indeed there is an execution of low priority
					job.preemtion( );
					JobsQueue.getInstance( ).pushFirstLowJob(job); // Put it back to the queue for later execution
				}
				highPriority = true;
			} else {
				if ( job == null || job.isFinished( ))
					$ = JobsQueue.getInstance().popLowPriorityJob( ); // There is no high priority jobs here
			}
			if ( $ == null) {
				// In this case there are two options 
				// I. There is no jobs to do in queue
				// II. There is low priority job is still running.
				try {
					Thread.sleep( 100); // Need to sleep before starting new round
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				continue;
			} else {
				$.addStopListener(this);
				$.addRejectListener( this);
				new Thread( job = $).start( ); // Execute new job
			}
		}
		endpoint.stop( );
		channel.close( ); // Close JGroups channel
	}
	
	public void finish( ) {
		finished = true;
	}

	@Override
	public void notifyStopped( final ServerJob job) {
		highPriority = false;
		try {
			ClusterMessage message = new ClusterMessage( Util.objectToByteBuffer( job), ClusterMessageType.JOB_FINISHED);
			channel.send( null, null, Util.objectToByteBuffer( message));
			log.info( "Notify cluster the job " + job.getJobId() + " has been finished");
		} catch (Exception e) {
			log.error( e);
		}
	}

	@Override
	public void notifyRejected( final ServerJob job) {
		try {
			ClusterMessage message = new ClusterMessage( Util.objectToByteBuffer( job), ClusterMessageType.JOB_REDIRECT);
			Vector<Address> members = channel.getView().getMembers();
			Random rnd = new Random( );
			Address address = null;
			while( address == null) { // Need to randomly peek the new node to send the job there
				int index;
				if ( members.get( index = rnd.nextInt( members.size( ))) != channel.getAddress()) {
					address =  members.get( index);
					break;
				}
			}
			channel.send( address, channel.getAddress( ), Util.objectToByteBuffer( message));
			log.info( "Notify cluster the job " + job.getJobId() + " has to be redirected");
		} catch (Exception e) {
			log.error( e);
		}
	}
	
	public static void main(String[] args) {
		if ( args.length != 2) {
			System.err.println( "Wrong parameters.");
			System.err.println( "Expected: server <serverName> <serverEndpoint>");
			return;
		}
		
		String serverName = args[0];
		String serverEndpoint = args[1];
		try {
			Properties properties = new Properties( );
			properties.load( Server.class.getClassLoader().getResourceAsStream( "server.properties"));
			String dispatchAddress = properties.getProperty( "dispatcher");
			// Register the server at dispatcher
			DispatcherService service = new DispatcherService( new URL( dispatchAddress), new QName("http://core.project.cs236635.technion.il.ac/", "DispatcherService"));
			ServerInfo info = new ServerInfo( );
			info.setServerName( serverName);
			info.setServerEnpoint( serverEndpoint);
			service.getDispatcherPort().register( info);
		} catch ( Exception e) {
			e.printStackTrace();
			return;
		}
		
		Server server = new Server( serverName, serverEndpoint);
		new Thread( server).start( ); // Start serving
		
		BufferedReader reader = null;
		try {
			reader = new BufferedReader( new InputStreamReader( System.in));
			String line = "";
			while ( !line.equals( "exit")) {
				try {
					line = reader.readLine( ); // Just wait until finished or suitable to turn off.
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		} finally {
			try {
				reader.close( );
				server.finish( );
			} catch (IOException e) {
			}
		}
	}
}
